package org.zebra.redisson.queue;

import org.redisson.api.RBlockingQueue;
import org.zebra.redisson.util.RedissonUtil;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;

/**
 * 有界阻塞队列
 * <p>建议在非必要的情况下，都使用有界阻塞队列操作，避免队列未消费，导致redis内存暴增</p>
 * <p>redis属于轻量级分布式队列，如果需要更加健壮可靠的队列操作，请根据业务需求使用：RabbitMQ、RocketMQ、Kafka等专业队列中间件实现</p>
 *
 * @author yue
 */
public final class BoundedBlockingQueue<V> {

    private static final Map<String, BoundedBlockingQueue> boundedBlockingQueueCache = new ConcurrentHashMap<>();

    /**
     * https://github.com/redisson/redisson/issues/3020
     * RBoundedBlockingQueue有界阻塞队列有bug，capacity值只减不增，导致队列最终可用容量为0，无法继续插入数据
     */
    private RBlockingQueue<V> boundedBlockingQueue;

    private BoundedBlockingQueue(RBlockingQueue<V> boundedBlockingQueue) {
        this.boundedBlockingQueue = boundedBlockingQueue;
    }

    public RBlockingQueue<V> getBoundedBlockingQueue() {
        return boundedBlockingQueue;
    }

    /**
     * ======有界阻塞队列操作======
     * 建议在非必要的情况下，都使用有界阻塞队列操作，避免队列未消费，导致redis内存暴增
     * redis属于轻量级分布式队列，如果需要更加健壮可靠的队列操作，请根据业务需求使用：RabbitMQ、RocketMQ、Kafka等专业队列中间件实现
     */
    /**
     * 获得有界阻塞队列
     * <p>建议在非必要的情况下，都使用有界阻塞队列操作，避免队列未消费，导致redis内存暴增</p>
     * <p>redis属于轻量级分布式队列，如果需要更加健壮可靠的队列操作，请根据业务需求使用：RabbitMQ、RocketMQ、Kafka等专业队列中间件实现</p>
     */
    public static <V> BoundedBlockingQueue<V> get(String queueName) {
        /**
         * https://github.com/redisson/redisson/issues/3020
         * RBoundedBlockingQueue有界阻塞队列有bug，capacity值只减不增，导致队列最终可用容量为0，无法继续插入数据
         */
        BoundedBlockingQueue<V> boundedBlockingQueue = boundedBlockingQueueCache.get(queueName);
        if (boundedBlockingQueue == null) {
            boundedBlockingQueue =
                    new BoundedBlockingQueue<>(RedissonUtil.getClient().getBlockingQueue(queueName));
            boundedBlockingQueueCache.put(queueName, boundedBlockingQueue);
        }

        return boundedBlockingQueue;
    }

    /**
     * 添加有界阻塞队列值
     * <p>队列大小限制为100000，超过丢弃</p>
     *
     * @param data 值
     */
    public boolean add(V data) {
        return add(data, 100000);
    }

    /**
     * 添加list有界阻塞队列值
     *
     * @param data      值
     * @param blockSize 队列最大大小，队列满后，继续添加的数据将被直接丢弃
     */
    public boolean add(V data, int blockSize) {
        int size = boundedBlockingQueue.size();
        if (size >= blockSize) {
            return false;
        }
        return boundedBlockingQueue.offer(data);
    }

    public V consumer() {
        return boundedBlockingQueue.poll();
    }

    /**
     * 获取并删除有界阻塞队列值
     *
     * @param consumeSize 一次消费多少条数据
     */
    public List<V> consumer(int consumeSize) {
        return boundedBlockingQueue.poll(consumeSize);
    }

    /**
     * 有界队列删除队列数据
     */
    public boolean remove(V data) {
        return boundedBlockingQueue.remove(data);
    }

    public boolean destroy() {
        return boundedBlockingQueue.delete();
    }

    /**
     * 订阅阻塞队列
     */
    public int consumer(Consumer<V> consumer) {
        return boundedBlockingQueue.subscribeOnElements(consumer);
    }
}
